package main

import (
	"log"
	"io"
	"encoding/json"
	"net/http"
	// "os"
	"fmt"
	"sync"
)

type WorkerPool struct {
	workers []*Worker
	pool chan *Worker
}

func NewWorkerPool(maxWorkers int) *WorkerPool {
	p := &WorkerPool {
		workers: make([]*Worker, maxWorkers),
		pool: make(chan *Worker, maxWorkers),
	}

	for i, _ := range p.workers {
		worker := NewWorker(0)
		p.workers[i] = worker
		p.pool <- worker
	}

	return p
}

func (p *WorkerPool) Start() {
	for _, worker := range p.workers {
		worker.Start()
	}
}

func (p *WorkerPool) Stop() {
	for _, worker := range p.workers {
		worker.Stop()
	}
}

func (p *WorkerPool) Get() *Worker {
	return <-p.pool
}

func (p *WorkerPool) Put(w *Worker) {
	p.pool <- w
}

type Service struct {
	workers *WorkerPool
	jobs chan interface{}
	maxJobs int
	wg sync.WaitGroup
}

func NewService(maxWorkers, maxJobs int) *Service {
	return &Service {
		workers: NewWorkerPool(maxWorkers),
		jobs: make(chan interface{}, maxJobs),
	}
}

func (p *Service) Start() {
	p.jobs = make(chan interface{}, p.maxJobs)

	p.wg.Add(1)
	p.workers.Start()

	go func() {
		defer p.wg.Done()

		for job := range p.jobs{
			go func(job interface{}) {
				worker := p.workers.Get()

				defer p.workers.Put(worker)

				worker.AddJob(job)
			}(job)
		}
	}()
}

func (p *Service) Stop() {
	p.workers.Stop()
	close(p.jobs)
	p.wg.Wait()
}


func (p *Service) AddJob(job interface{}) {
	p.jobs <- job
}


type Worker struct {
	job chan interface{}
	quit chan bool
	wg sync.WaitGroup
}

func NewWorker(maxJobs int) *Worker {
	return &Worker {
		job: make(chan interface{}, maxJobs),
		quit: make(chan bool),
	}
}

func (p *Worker) Start() {
	p.wg.Add(1)

	go func() {
		defer p.wg.Done()

		for {
			select {
			case job := <-p.job:
				fmt.Println(job)

			case <-p.quit:
				return
			}
		}
	}()
}

func (p *Worker) Stop() {
	p.quit <- true
	p.wg.Wait()
}

func (p *Worker) AddJob(job interface{}) {
	p.job <- job
}

var (
	// MaxWorker = os.Getenv("MAX_WORKERS")
	// MAXQueue = os.Getenv("MAX_QUEUE")
	MaxWorker = 500
	MaxQueue = 1000
	MaxLength = int64(50000)
)

func main() {
	service := NewService(MaxWorker, MaxQueue)

	service.Start()
	defer service.Stop()

	http.HandleFunc("/jobs", func(w http.ResponseWriter, r *http.Request) {
		if r.Method != "POST" {
			w.WriteHeader(http.StatusMethodNotAllowed)
			return
		}

		var jobs []interface{}

		err := json.NewDecoder(io.LimitReader(r.Body, MaxLength)).Decode(&jobs)

		if err != nil {
			w.Header().Set("Content-Type", "application/json; charset=UTF-8")
			w.WriteHeader(http.StatusBadRequest)
			return
		}

		for _, job := range jobs {
			service.AddJob(job)
		}

		w.WriteHeader(http.StatusOK)
	})

	log.Fatal(http.ListenAndServe(":8080", nil))
}